package com.htdata.iiot.api.tsdb.utils;

import com.htdata.iiot.api.hbase.pojo.DelMetric;
import com.htdata.iiot.api.hbase.utils.HbaseOperation;
import com.htdata.iiot.api.tsdb.config.Configs;
import net.opentsdb.core.Const;
import org.apache.hadoop.hbase.client.Result;
import org.apache.log4j.Logger;
import org.hbase.async.Bytes;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;

/**
 * @author chensi
 */
public class TsdbTools {
    private static Logger LOGGER = Logger.getLogger(TsdbTools.class);
    public static final short FLAG_BITS = 4;
    public static final short FLAG_FLOAT = 0x8;
    static short flags = Const.FLAG_FLOAT | 0x3;  // A float stored on 4 bytes.

    /**
     * tsdb表中列
     * @param timestamp
     * @param flags
     * @return
     */
    public static byte[] buildQualifier(final long timestamp, final short flags) {
        final long base_time;
        if ((timestamp & Const.SECOND_MASK) != 0) {
            // drop the ms timestamp to seconds to calculate the base timestamp
            base_time = ((timestamp / 1000) - ((timestamp / 1000)
                    % Const.MAX_TIMESPAN));
            final int qual = (int) (((timestamp - (base_time * 1000)
                    << (Const.MS_FLAG_BITS)) | flags) | Const.MS_FLAG);
            return Bytes.fromInt(qual);
        } else {
            base_time = (timestamp - (timestamp % Const.MAX_TIMESPAN));
            final short qual = (short) ((timestamp - base_time) << Const.FLAG_BITS
                    | flags);
            return Bytes.fromShort(qual);
        }
    }

    /**
     * tsdb表rowkey中的timestamp
     * @param timestamp
     * @return
     */
    public static byte[] getRowKeyTime(long timestamp){
        long base_time = 0;
        if ((timestamp & Const.SECOND_MASK) != 0) {
            // drop the ms timestamp to seconds to calculate the base timestamp
            base_time = ((timestamp / 1000) -
                    ((timestamp / 1000) % Const.MAX_TIMESPAN));
        } else {
            base_time = (timestamp - (timestamp % Const.MAX_TIMESPAN));
        }
        int tmp = (int) base_time;
        byte[] b = new byte[4];
        b[0] = (byte)(tmp >>> 24);
        b[1] = (byte)(tmp >>> 16);
        b[2] = (byte)(tmp >>> 8);
        b[3] = (byte)(tmp >>> 0);
        return b;
    }

    /**
     * 单个查询
     * 通过metric查询row_key
     * @param metric
     * @param time
     * @param tid
     * @param did
     * @return
     * @throws Exception
     */
    public static byte[] getRowKeyFromMetric(String metric,String time,String tid,String did) throws Exception {
        String tagk1 = "tid";
        String tagv1 = tid;
        String tagk2 = "did";
        String tagv2 = did;
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date = sdf.parse(time);
        byte[] tms = getRowKeyTime(date.getTime());
        String[] rowKeys = new String[5];
        rowKeys[0] = metric;
        rowKeys[1] = tagk1;
        rowKeys[2] = tagv1;
        rowKeys[3] = tagk2;
        rowKeys[4] = tagv2;
        Result[] results = HbaseOperation.getResults(Configs.TSDBConfig.UID_TABLE, rowKeys);
        byte[] m = results[0].getValue("id".getBytes(), "metrics".getBytes());
        if(m==null){
            return null;
        }
        byte[] tagk1B = results[1].getValue("id".getBytes(), "tagk".getBytes());
        byte[] tagv1B = results[2].getValue("id".getBytes(), "tagv".getBytes());
        byte[] tagk2B = results[3].getValue("id".getBytes(), "tagk".getBytes());
        byte[] tagv2B = results[4].getValue("id".getBytes(), "tagv".getBytes());
        byte[] row_key = Tools.combineArray(m, tms, tagk1B, tagv1B, tagk2B, tagv2B);
        byte[] lastest_key = prefixKeyWithSalt(row_key);
        return lastest_key;
    }

    /**
     * list大小应控制在
     * @param list
     * @return
     * @throws Exception
     */
    public static List<byte[]> getRowKeyFromMetric(List<DelMetric> list) throws Exception {
        String rowKeys[] = new String[list.size()*5];
        int index = 0;
        for(DelMetric m:list){
            rowKeys[index++] = m.getMetric();
            rowKeys[index++] = "tid";
            rowKeys[index++] = m.getTid();
            rowKeys[index++] = "did";
            rowKeys[index++] = m.getDid();
        }
        Result[] results = HbaseOperation.getResults(Configs.TSDBConfig.UID_TABLE, rowKeys);
        List<byte[]> rowKeyList = new ArrayList<>();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        index=0;
        int count = 0;
        while(index<results.length){
            byte[] m = results[index++].getValue("id".getBytes(),"metrics".getBytes());
            byte[] tagk1 = results[index++].getValue("id".getBytes(),"tagk".getBytes());
            byte[] tagv1 = results[index++].getValue("id".getBytes(), "tagv".getBytes());
            byte[] tagk2 = results[index++].getValue("id".getBytes(),"tagk".getBytes());
            byte[] tagv2 = results[index++].getValue("id".getBytes(), "tagv".getBytes());
            String[] times = list.get(count).getTime();
            //后期可以进一步优化返回不包含日期的rowkey,时间返回之后再进行插入合成
            for (String time:times){
                byte[] tms = getRowKeyTime(sdf.parse(time).getTime());
                byte[] row_key = Tools.combineArray(m, tms, tagk1, tagv1, tagk2, tagv2);
                if(row_key!=null) {
                    byte[] lastest_key = prefixKeyWithSalt(row_key);
                    rowKeyList.add(lastest_key);
                }
            }
        }
        return rowKeyList;
    }

    public static List<byte[]> getRowKeyFromMetricRefact(List<DelMetric> list) throws Exception {
        String rowKeys[] = new String[list.size()*5];
        int index = 0;
        for(DelMetric m:list){
            rowKeys[index++] = m.getMetric();
            rowKeys[index++] = "tid";
            rowKeys[index++] = m.getTid();
            rowKeys[index++] = "did";
            rowKeys[index++] = m.getDid();
        }
        Result[] results = HbaseOperation.getResults(Configs.TSDBConfig.UID_TABLE, rowKeys);
        List<byte[]> rowKeyList = new ArrayList<>();
        index=0;
        while(index<results.length){
            byte[] m = results[index++].getValue("id".getBytes(),"metrics".getBytes());
            byte[] tagk1 = results[index++].getValue("id".getBytes(),"tagk".getBytes());
            byte[] tagv1 = results[index++].getValue("id".getBytes(), "tagv".getBytes());
            byte[] tagk2 = results[index++].getValue("id".getBytes(),"tagk".getBytes());
            byte[] tagv2 = results[index++].getValue("id".getBytes(), "tagv".getBytes());
            byte[] tms = new byte[Configs.TSDBConfig.TIMESTAMP_BYTES];
            byte[] row_key = Tools.combineArray(m, tms,tagk1, tagv1, tagk2, tagv2);
            if(row_key!=null) {
                byte[] lastest_key = prefixKeyWithSalt(row_key);
                rowKeyList.add(lastest_key);
            }
        }
        return rowKeyList;
    }

    /**
     * 将未添加盐值的row-key添加盐值并返回
     * 注意:Configs.TSDBConfig为TSDB的配置，请与服务器TSDB的配置相对应
     * @param row_key 未添加盐值的row-key
     * @return 已经添加好盐值的row-key
     */
    public static byte[] prefixKeyWithSalt(final byte[] row_key){
        if(Configs.TSDBConfig.SALT_WIDTH>0){
            final int tags_start = Configs.TSDBConfig.METRIC_WIDTH + Configs.TSDBConfig.TIMESTAMP_BYTES;
            // we want the metric and tags, not the timestamp
            final byte[] salt_base = new byte[row_key.length - Configs.TSDBConfig.TIMESTAMP_BYTES];
            System.arraycopy(row_key,0,salt_base,0,Configs.TSDBConfig.METRIC_WIDTH);
            System.arraycopy(row_key,tags_start,salt_base,Configs.TSDBConfig.METRIC_WIDTH,row_key.length - Configs.TSDBConfig.METRIC_WIDTH - Configs.TSDBConfig.TIMESTAMP_BYTES);
            int modulo = Arrays.hashCode(salt_base) % Configs.TSDBConfig.SALT_BUCKETS;
            if (modulo < 0) {
                // make sure we return a positive salt.
                modulo = modulo * -1;
            }
            final byte[] bytes = new byte[Configs.TSDBConfig.SALT_WIDTH];
            int shift = 0;
            for (int i = 1;i <= Configs.TSDBConfig.SALT_WIDTH; i++) {
                bytes[Configs.TSDBConfig.SALT_WIDTH - i] = (byte) (modulo >>> shift);
                shift += 8;
            }
            byte[] result = new byte[row_key.length + Configs.TSDBConfig.SALT_WIDTH];
            System.arraycopy(bytes,0,result,0,bytes.length);
            System.arraycopy(row_key,0,result,bytes.length,row_key.length);
            return result;
        }
        return null;
    }

}
